package com.xunkids.bos.migration.app.configuration;


import com.xunkids.bos.migration.business.service.bos.impl.DefaultKafkaSubscriber;
import com.xunkids.bos.migration.component.transmission.api.QueuePublisher;
import com.xunkids.bos.migration.component.transmission.api.QueueSubscriber;
import com.xunkids.bos.migration.component.log.LogService;
import com.xunkids.bos.migration.component.transmission.kafka.DefaultKafkaPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Locale;

/**
 * common queue configuration
 *
 * @author xigexb
 * @version 1.0.0
 * @since 2022/11/4 10:49
 */
@Configuration
@ConfigurationPropertiesScan(basePackages = {"com.xunkids.bos.migration.app.configuration"})
public class CommonQueueConfiguration {

    private static final Logger logger = LoggerFactory.getLogger("bosDataMigrationSys");

    @ConditionalOnMissingBean(value = {QueuePublisher.class})
    @Bean("queuePublisher")
    public QueuePublisher queuePublisher(SysQueueProperties properties, @Qualifier("logService") LogService logService) {
        QueuePublisher publisher = null;
        switch (properties.getType().toLowerCase(Locale.ROOT)) {
            case "kafka":
                publisher = new DefaultKafkaPublisher(new HashMap<>(properties.getConfigKV()), logService);
                break;

        }
        if (logger.isInfoEnabled()) {
            logger.info("init common queue publisher success");
        }
        return publisher;
    }


    @ConditionalOnMissingBean(value = {QueueSubscriber.class})
    @Bean("queueSubscriber")
    public QueueSubscriber queueSubscriber(SysQueueProperties properties) {
        QueueSubscriber subscriber = null;
        switch (properties.getType().toLowerCase(Locale.ROOT)) {
            case "kafka":
                subscriber = new DefaultKafkaSubscriber(new HashMap<>(properties.getConfigKV()));
                break;

        }
        if (logger.isInfoEnabled()) {
            logger.info("init common queue subscriber success");
        }
        return subscriber;
    }


}
